package com.pie4cloud.pie.job.executor.kaf;

import com.alibaba.fastjson.JSON;
import com.pie4cloud.pie.bus.api.consumer.IKfkaConsumer;
import com.pie4cloud.pie.bus.api.consumer.annotation.KfkConsumer;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.job.api.entity.GatherMessageEntity;
import com.pie4cloud.pie.job.executor.dto.StatisticalRequest;
import com.pie4cloud.pie.job.executor.enums.StatisticalType;
import com.pie4cloud.pie.job.executor.mapper.CommonRepository;
import com.pie4cloud.pie.job.executor.services.StatisticalService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Map;

@Slf4j
@Service
@KfkConsumer(handler = "gatherKakHandler")
public class DataPersistenceKafkaHandler implements IKfkaConsumer {

	@Resource
	private CommonRepository commonRepository;
	@Resource
	private StatisticalService statisticalService;

	@Override
	public R consume(String data,String topic,String groupId) throws Exception {
		try {
			GatherMessageEntity gatherMessageEntity = JSON.parseObject(data, GatherMessageEntity.class);
			String tableName = gatherMessageEntity.getTableName();
			Map<String, String> decodeResult = gatherMessageEntity.getDecodeResult();
			StatisticalRequest statisticalRequest=new StatisticalRequest();
			statisticalRequest.setStatisticalType(StatisticalType.DECODE);
			statisticalRequest.setPlaybillId(gatherMessageEntity.getPlayBillId());
			statisticalRequest.setFilePath(gatherMessageEntity.getLocalFilepath());
			statisticalRequest.setWhenTime(gatherMessageEntity.getTimeLevel());
			if(statisticalService.check(statisticalRequest)){
				if (StringUtils.isNotBlank(getTableName(tableName)) && decodeResult != null && !decodeResult.isEmpty()) {
					//commonRepository.InsertOneRecored(decodeResult,gatherMessageEntity.getTableName());
					statisticalService.doOperation(statisticalRequest);
				}
			}


		} catch (Exception e) {
			return R.failed("insert data fail.");
		}

		return R.ok();
	}

	private static String getTableName(String oldTableName) {
		if (StringUtils.contains(oldTableName,".")) {
			return StringUtils.split(oldTableName,".")[1];
		}
		return oldTableName;
	}
}
